package com.gitee.zhziyun.cnfsystem.utils.FileUtil;

import com.gitee.zhziyun.cnfsystem.utils.XXHashUtil;

import java.io.*;
import java.net.Socket;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

/**
 * <p>多线程文件发送工具（支持断点续传、进度控制）<p/>
 * <p></p>
 * <p>功能特性：</p>
 * <p>1. 自动根据文件大小选择单线程/多线程传输</p>
 * <p>2. 实时传输进度回调</p>
 * <p>3. 支持暂停/恢复/取消操作</p>
 * <p>4. 断点续传支持</p>
 * <p>5. 大文件分块校验</p>
 */
public class ThreadedFileSender {
    // 常量定义
    private static final int BLOCK_SIZE = 1024 * 4; // 4KB数据块大小
    private static final long MULTI_THREAD_THRESHOLD = 500 * 1024; // 500KB阈值
    private static final int THREAD_POOL_SIZE = 4; // 线程池大小

    // 传输控制状态
    private final AtomicBoolean paused = new AtomicBoolean(false);
    private final AtomicBoolean cancelled = new AtomicBoolean(false);

    // 传输元数据
    private String hostIp;
    private int port;
    private final String filePath;
    private final ProgressListener progressListener;
    private long transferredBytes = 0;
    private long totalBytes;

    /**
     * 进度监听接口
     */
    public interface ProgressListener {
        default void onProgress(long current, long total) { }
        default void onPaused() { }
        default void onResumed() { }
        default void onCompleted() { }
        default void onError(Exception e){ }
        default void onCancelled() { }
    }

    /**
     * 构造方法
     * @param filePath 待发送文件路径
     * @param listener 进度监听器
     */
    public ThreadedFileSender(String filePath, ProgressListener listener) {
        this.filePath = filePath;
        this.progressListener = listener;
    }

    // 主要控制方法
    public void start(String hostIp, int port) {
        this.hostIp = hostIp;
        this.port = port;
        new Thread(this::doTransfer).start();
    }
    public void pause() { paused.set(true); progressListener.onPaused(); }
    public void resume() { paused.set(false); synchronized (paused) { paused.notifyAll(); } progressListener.onResumed(); }
    public void cancel() { cancelled.set(true); resume(); }

    /**
     * 核心传输方法
     */
    private void doTransfer() {
        try {
            File file = new File(filePath);
            totalBytes = file.length();

            // 记录传输状态（用于断点续传）
            ResumableFileTransfer.saveTransferStatus(filePath, 0);

            if (totalBytes > MULTI_THREAD_THRESHOLD) {
                multiThreadTransfer(file);
            } else {
                singleThreadTransfer(file);
            }

            // 传输完成清理状态
            ResumableFileTransfer.clearTransferStatus(filePath);
            progressListener.onCompleted();
        } catch (Exception e) {
            progressListener.onError(e);
        }
    }

    /**
     * 单线程传输实现
     */
    private void singleThreadTransfer(File file) throws IOException {
        try (Socket socket = new Socket(hostIp, port);
             FileInputStream fis = new FileInputStream(file);
             DataOutputStream dos = new DataOutputStream(socket.getOutputStream())) {

            sendFileMetadata(dos, file);

            byte[] buffer = new byte[BLOCK_SIZE];
            int bytesRead;

            while (!cancelled.get() && (bytesRead = fis.read(buffer)) != -1) {
                try {
                    waitIfPaused();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("Thread interrupted during file transfer", e);
                }
                dos.write(buffer, 0, bytesRead);
                updateProgress(bytesRead);

                // 更新断点位置
                ResumableFileTransfer.saveTransferStatus(filePath, transferredBytes);
            }
        }
    }

    /**
     * 多线程传输实现
     */
    private void multiThreadTransfer(File file) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        CompletionService<Void> completionService = new ExecutorCompletionService<>(executor);

        try (Socket socket = new Socket(hostIp, port);
             DataOutputStream dos = new DataOutputStream(socket.getOutputStream())) {

            sendFileMetadata(dos, file);

            long chunkCount = (file.length() + BLOCK_SIZE - 1) / BLOCK_SIZE;
            AtomicLong counter = new AtomicLong(0);

            for (long i = 0; i < chunkCount; i++) {
                final long chunkIndex = i;
                completionService.submit(() -> {
                    if (cancelled.get()) return null;

                    try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
                        long position = chunkIndex * BLOCK_SIZE;
                        int chunkSize = (int) Math.min(BLOCK_SIZE, file.length() - position);
                        byte[] buffer = new byte[chunkSize];

                        raf.seek(position);
                        raf.readFully(buffer);

                        waitIfPaused();

                        // 计算分块哈希
                        byte[] chunkHash = XXHashUtil.calculateHash(buffer);

                        synchronized (dos) {
                            dos.writeLong(chunkIndex);
                            dos.writeInt(chunkSize);
                            dos.write(chunkHash);  // 发送原始字节形式的哈希
                            dos.write(buffer);
                        }

                        // 使用精确的字节数更新进度
                        updateProgress(chunkSize);
                        long currentCounter = counter.incrementAndGet();

                        // 每完成1%或10%保存一次进度
                        if (currentCounter % Math.max(1, chunkCount/100) == 0) {
                            ResumableFileTransfer.saveTransferStatus(filePath,
                                    chunkIndex * BLOCK_SIZE);
                        }
                    }
                    return null;
                });
            }

            for (long i = 0; i < chunkCount; i++) {
                completionService.take().get();
                if (cancelled.get()) break;
            }
        } finally {
            executor.shutdownNow();
        }
    }

    // 辅助方法
    private void sendFileMetadata(DataOutputStream dos, File file) throws IOException {
        String fileHash = XXHashUtil.calculateFileHashToString(file);
        dos.writeUTF(fileHash);
        dos.writeUTF(file.getName());
        dos.writeLong(file.length());
        dos.writeBoolean(totalBytes > MULTI_THREAD_THRESHOLD);
    }

    private void waitIfPaused() throws InterruptedException {
        while (paused.get() && !cancelled.get()) {
            synchronized (paused) {
                paused.wait();
            }
        }
    }

    private void updateProgress(int bytes) {
        transferredBytes += bytes;
        progressListener.onProgress(transferredBytes, totalBytes);
    }
}